/*
 *   Copyright 1999-2016 Asiainfo Technologies(China),Inc.
 *
 *   Licensed under the Apache License, Version 2.0 (the "License");
 *   you may not use this file except in compliance with the License.
 *   You may obtain a copy of the License at
 *
 *        http://www.apache.org/licenses/LICENSE-2.0
 *
 *   Unless required by applicable law or agreed to in writing, software
 *   distributed under the License is distributed on an "AS IS" BASIS,
 *   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 *   See the License for the specific language governing permissions and
 *   limitations under the License.
 *
 */

package com.asiainfo.redis.service;

import com.asiainfo.redis.entity.BusiDataChange;
import com.asiainfo.redis.entity.BusiDataRule;
import com.asiainfo.redis.entity.BusiType;
import com.asiainfo.redis.entity.Cluster;
import com.asiainfo.redis.entity.ClusterBusiTypeRef;
import com.asiainfo.redis.repository.BusiDataChangeRepository;
import com.asiainfo.redis.repository.BusiDataRuleRepository;
import com.asiainfo.redis.repository.BusiTypeRepository;
import com.asiainfo.redis.repository.ClusterBusiTypeRefRepository;
import com.asiainfo.redis.repository.ClusterRepository;
import com.asiainfo.redis.scheduling.RefreshAllBusiTypeDataJob;
import com.asiainfo.redis.service.clusterhandler.RedisClusterHandler;
import org.apache.commons.lang.StringUtils;
import org.quartz.CronTrigger;
import org.quartz.Scheduler;
import org.quartz.Trigger;
import org.quartz.impl.triggers.CronTriggerImpl;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.data.redis.connection.RedisClusterConnection;
import org.springframework.data.redis.connection.RedisClusterNode;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.util.Assert;
import redis.clients.jedis.JedisCluster;
import redis.clients.util.SafeEncoder;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.io.IOException;
import java.net.InetAddress;
import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.concurrent.Callable;
import java.util.concurrent.CompletionService;
import java.util.concurrent.ExecutorCompletionService;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

/**
 * @version v 1.0 on 2016/8/10 16:33
 * @auther william.xu
 */
@Service
public class ScheduleService {

    protected static final Logger logger = LoggerFactory.getLogger(ScheduleService.class);

    @Resource
    private ClusterBusiTypeRefRepository clusterBusiTypeRefRepository;

    @Resource
    private BusiDataChangeRepository busiDataChangeRepository;

    @Resource
    private BusiDataRuleRepository busiDataRuleRepository;

    @Resource
    private ClusterRepository clusterRepository;

    @Resource
    private BusiTypeRepository busiTypeRepository;

    @Resource
    private BusiConfigService busiConfigService;

    @Autowired
    @Qualifier("schedulerFactory")
    private Scheduler scheduler;


    @Resource
    private JdbcTemplate jdbcTemplate;

    @Resource
    private CronTrigger refreshChangeBusiDataCronTrigger;

    @Resource
    private CronTrigger refreshAllBusiTypeDataCronTrigger;

    public static final Integer STATUS_PENDING = 0;
    public static final Integer STATUS_RUNNING = 1;
    public static final Integer STATUS_COMPLETED = 2;
    public static final Integer STATUS_FAILED = 3;

    public static final Integer MAX_THREADS = 80;


    public List<ClusterBusiTypeRef> getClusterBusiTypeList(){
        List<ClusterBusiTypeRef> result = new ArrayList<>();
        List<ClusterBusiTypeRef> list = clusterBusiTypeRefRepository.findAllByOrderByClusterIdAsc();
        Long oldClusterId = null;
        for (ClusterBusiTypeRef clusterBusiTypeRef : list){
            Long clusterId = clusterBusiTypeRef.getClusterId();
            Long currentClusterId = clusterBusiTypeRef.getCurrentClusterId();
            Cluster cluster = clusterRepository.findOne(clusterId);
            Assert.notNull(cluster);
            clusterBusiTypeRef.setClusterName(cluster.getClusterName());
            if (currentClusterId != null){
                Cluster currentCluster = clusterRepository.findOne(currentClusterId);
                Assert.notNull(currentCluster);
                clusterBusiTypeRef.setCurrentClusterName(currentCluster.getClusterName());
            }

            BusiType busiType = busiTypeRepository.findOne(clusterBusiTypeRef.getBusiTypeId());
            Assert.notNull(busiType);
            clusterBusiTypeRef.setBusiCode(busiType.getBusiCode());
            clusterBusiTypeRef.setBusiName(busiType.getBusiName());
            if (!clusterId.equals(oldClusterId)){
                ClusterBusiTypeRef row = new ClusterBusiTypeRef();
                row.setClusterId(clusterId);
                row.setClusterName(clusterBusiTypeRef.getClusterName());
                result.add(row);
                oldClusterId = clusterId;
            }
            result.add(clusterBusiTypeRef);
        }
        return result;

    }

    public String getRefreshChangeBusiDataCronExpression() throws Exception{
        CronTriggerImpl trigger = (CronTriggerImpl)refreshChangeBusiDataCronTrigger;
        Trigger exists = scheduler.getTrigger(trigger.getKey());
        if (exists == null){
            //return "0 0/1 * * * ?";
            return "";
        }else{
            return ((CronTriggerImpl)exists).getCronExpression();
        }

    }

    public void setRefreshChangeBusiDataCronExpression(String cronExpression) throws Exception{
        CronTriggerImpl trigger = (CronTriggerImpl)refreshChangeBusiDataCronTrigger;
        trigger.setStartTime(new Date());
        trigger.setEndTime(null);
        trigger.setCronExpression(cronExpression);
        //trigger.setName("refreshChangeBusiDataCronTrigger");//这里默认是注入的bean名称
        Trigger exits = scheduler.getTrigger(trigger.getKey());
        if (exits == null) {
            scheduler.scheduleJob(trigger);
        } else {
            scheduler.rescheduleJob(trigger.getKey(),trigger);
        }

    }

    public List<BusiDataChange> getNotHandleBusiDataChangeList(){
        return busiDataChangeRepository.findFirst1000ByStatus(0);
    }

    @Transactional(rollbackFor = Exception.class)
    public void saveBusiDataChange(BusiDataChange busiDataChange){
        if (busiDataChange.getStatus().equals(ScheduleService.STATUS_RUNNING)) {
            busiDataChangeRepository.save(busiDataChange);
        }else{
            //把记录移到redis_busi_data_change_his表
            jdbcTemplate.update("insert into redis_busi_data_change_his values(?,?,?,?,?,?,?)",
                    new Object[]{busiDataChange.getId(), busiDataChange.getBusiDataRuleId(),
                            busiDataChange.getSqlCondition(), busiDataChange.getOperFlag(),
                            busiDataChange.getStatus(), busiDataChange.getErrorMsg(),
                            busiDataChange.getUpdateTime()});
            jdbcTemplate.update("delete from redis_busi_data_change where id=? ",
                    new Object[]{busiDataChange.getId()});
        }
    }

    public ClusterBusiTypeRef getClusterBusiTypeRef(Long id){
        ClusterBusiTypeRef clusterBusiTypeRef = clusterBusiTypeRefRepository.findOne(id);
        Assert.notNull(clusterBusiTypeRef, "该ClusterBusiTypeRefId找不到，id="+id);
        return clusterBusiTypeRef;
    }

    public String getRefreshAllBusiDataCronExpression(Long clusterBusiTypeRefId) throws Exception{

        ClusterBusiTypeRef clusterBusiTypeRef = getClusterBusiTypeRef(clusterBusiTypeRefId);

        CronTriggerImpl trigger = (CronTriggerImpl)refreshAllBusiTypeDataCronTrigger;
        trigger.setName("refreshAllBusiTypeDataCronTrigger_" + clusterBusiTypeRef.getId());

        Trigger exists = scheduler.getTrigger(trigger.getKey());
        if (exists == null){
            //return "0 0/1 * * * ?";
            return "";
        }else{
            return ((CronTriggerImpl)exists).getCronExpression();
        }

    }

    public void setRefreshAllBusiDataCronExpression(Long clusterBusiTypeRefId, String cronExpression)
            throws Exception{
        ClusterBusiTypeRef clusterBusiTypeRef = getClusterBusiTypeRef(clusterBusiTypeRefId);

        CronTriggerImpl trigger = (CronTriggerImpl)refreshAllBusiTypeDataCronTrigger;
        trigger.setStartTime(new Date());
        trigger.setEndTime(null);
        trigger.setCronExpression(cronExpression);

        trigger.getJobDataMap().put("ID", clusterBusiTypeRefId);
        trigger.setName("refreshAllBusiTypeDataCronTrigger_" + clusterBusiTypeRef.getId());

        Trigger exits = scheduler.getTrigger(trigger.getKey());
        if (exits == null) {
            scheduler.scheduleJob(trigger);
        } else {
            scheduler.rescheduleJob(trigger.getKey(),trigger);
        }

    }

    public void refreshAllData(Long clusterBusiTypeRefId) throws Exception{
        final ClusterBusiTypeRef clusterBusiTypeRef = getClusterBusiTypeRef(clusterBusiTypeRefId);

        Cluster cluster = clusterRepository.findOne(clusterBusiTypeRef.getClusterId());
        Assert.notNull(cluster);
        final BusiType busiType = busiTypeRepository.findOne(clusterBusiTypeRef.getBusiTypeId());
        Assert.notNull(busiType);

        scheduleService.checkForRefreshData(clusterBusiTypeRef);

        new Thread(new Runnable() {
            @Override
            public void run() {
                try {
                    scheduleService.refreshAllData(clusterBusiTypeRef);
                }catch (Exception e){
                    logger.error("刷新数据失败：", e);
                }
            }
        }).start();

    }


    public void checkForRefreshData(ClusterBusiTypeRef clusterBusiTypeRef) throws Exception{
        Cluster cluster = clusterRepository.findOne(clusterBusiTypeRef.getClusterId());
        Assert.notNull(cluster);
        if (cluster.getStandbyId() == null){
            throw new Exception("此集群没有配置备用集群，不能全量更新！");
        }
        if (clusterBusiTypeRef.getRefreshStatus() == STATUS_RUNNING){
            throw new Exception("该业务正在全量更新数据，请等待更新完成！");
        }
    }

    public void refreshAllData(ClusterBusiTypeRef clusterBusiTypeRef) throws Exception{
        Long refreshClusterId = null;

        Long currentClusterId = clusterBusiTypeRef.getCurrentClusterId();
        if (currentClusterId == null){
            currentClusterId = clusterBusiTypeRef.getClusterId();
        }
        Cluster cluster = clusterRepository.findOne(currentClusterId);
        Assert.notNull(cluster);
        if (cluster.getType() == 0){
            //当前是主用集群时-----------------
            //1.获取备用集群id
            refreshClusterId = cluster.getStandbyId();
            //2.开始刷备用集群，并设置刷新状态等信息
            doRefreshAllData(clusterBusiTypeRef, refreshClusterId, true);


            //3.开始刷主用集群
            refreshClusterId = cluster.getId();
            doRefreshAllData(clusterBusiTypeRef, refreshClusterId, true);

        }else{
            //当前是备用集群时------------------
            //1.获取主用集群id
            Cluster masterCluster = clusterRepository.findByStandbyId(cluster.getId());
            Assert.notNull(masterCluster);
            //2.开始刷主用集群
            refreshClusterId = masterCluster.getId();
            doRefreshAllData(clusterBusiTypeRef, refreshClusterId, true);

            //3.开始刷备用集群
            refreshClusterId = cluster.getId();
            doRefreshAllData(clusterBusiTypeRef, refreshClusterId, false);//为了在正常情况下当前集群都保证是主用集群，所以这时刷完备用集群就不用通知了

        }

    }


    private String getHostName() throws Exception{
        return InetAddress.getLocalHost().getHostName();
    }

    @Resource
    private ScheduleService scheduleService;
    //如果在刷新数据过程中进程被kill掉，会导致当前refreshStatus一直为刷新中
    //这个方法在进程重启时，找到之前执行的记录，状态置为失败
    @PostConstruct
    public void init() throws Exception{
        scheduleService.resetRefreshStatus();
    }

    @Transactional(rollbackFor = Exception.class)
    public void resetRefreshStatus() throws Exception{
        clusterBusiTypeRefRepository.resetRefreshStatusByHostRunOn("由于在全量更新数据过程中进程被kill掉，更新失败！", getHostName());
    }

    private void doRefreshAllData(ClusterBusiTypeRef clusterBusiTypeRef, final Long toClusterId, boolean isPublishMsg)
            throws Exception {
        if (clusterBusiTypeRef.getRefreshStatus() == STATUS_RUNNING){
            throw new Exception("该业务正在全量更新数据，请等待更新完成！");
        }
        try {
            //设置正在刷新状态等信息
            clusterBusiTypeRef.setRefreshStatus(STATUS_RUNNING);
            clusterBusiTypeRef.setRefreshErrorMsg("");
            clusterBusiTypeRef.setRefreshStartTime(new Date());
            clusterBusiTypeRef.setRefreshEndTime(null);
            //quartz也是使用本地主机名生成唯一实例，org.quartz.simpl.SimpleInstanceIdGenerator.generateInstanceId
            clusterBusiTypeRef.setHostRunOn(getHostName());
            clusterBusiTypeRefRepository.save(clusterBusiTypeRef);

            List<BusiDataRule> list = busiDataRuleRepository.findByClusterIdAndBusiTypeIdAndScheduleFlag
                    (clusterBusiTypeRef.getClusterId(), clusterBusiTypeRef.getBusiTypeId(), 0);

            //spring jedis api调用keys方法是异步调用，跟多线程全量刷新会冲突，所以必须单线程调用
            logger.info("正在对Redis集群（clusterId="+ toClusterId +"）清除业务数据（为了提升清除效率，直接删除redis数据文件，并重启）");
//            for (final BusiDataRule busiDataRule : list){
//                busiConfigService.removeAllRedisData(busiDataRule, toClusterId);
//            }
            busiConfigService.removeAllRedisData(toClusterId);

            logger.info("正在对Redis集群（clusterId="+ toClusterId +"）全量写入业务数据");
            ExecutorService es = Executors.newFixedThreadPool(MAX_THREADS);
            CompletionService<Map> cs = new ExecutorCompletionService(es);
            for (final BusiDataRule busiDataRule : list){

                cs.submit(new Callable<Map>() {
                    @Override
                    public Map call() throws Exception {
                        Map result = new HashMap();
                        try {
                            String sql = "select * from (" + busiDataRule.getSql() + ") where 1=1";
                            busiConfigService.writeDataToRedis(busiDataRule, sql, true, 0, toClusterId);
                            result.put("state", new Integer(0));
                            result.put("msg", "");
                        }catch (Exception e){
                            String errorMsg = "刷新全量数据，在处理业务规则id="+busiDataRule.getId()+"时出现错误：";
                            logger.error(errorMsg, e);
                            result.put("state", new Integer(1));
                            result.put("msg", errorMsg + e.toString());
                        }
                        return result;
                    }
                });

            }
            es.shutdown();
            for (final BusiDataRule busiDataRule : list){
                Map result = cs.take().get();
                Integer state = (Integer)result.get("state");
                String msg = (String)result.get("msg");
                //只要发现有一个业务规则全量刷入redis出现失败，即立即取消其它任务的执行
                if (state.equals(1)){
                    es.shutdownNow();
                    throw new Exception(msg);
                }
            }
            es.awaitTermination(Long.MAX_VALUE, TimeUnit.DAYS);


            if (isPublishMsg){
                BusiType busiType = busiTypeRepository.findOne(clusterBusiTypeRef.getBusiTypeId());
                publishMessage(toClusterId, busiType.getBusiCode());

                clusterBusiTypeRef.setCurrentClusterId(toClusterId);
            }
            //刷新完成，设置成功状态
            clusterBusiTypeRef.setRefreshStatus(STATUS_COMPLETED);
            clusterBusiTypeRef.setRefreshErrorMsg("");
            clusterBusiTypeRef.setRefreshEndTime(new Date());
            clusterBusiTypeRefRepository.save(clusterBusiTypeRef);

        }catch (Exception e){

            //刷新失败，设置失败状态
            clusterBusiTypeRef.setRefreshStatus(STATUS_FAILED);
            clusterBusiTypeRef.setRefreshErrorMsg(e.toString());
            clusterBusiTypeRef.setRefreshEndTime(new Date());
            clusterBusiTypeRefRepository.save(clusterBusiTypeRef);

            throw e;

        }
    }

    //刷新完成，发布消息通知，并把当前集群写入redis_current_cluster表
    private void publishMessage(Long clusterId, String busiCode) throws Exception{
        RedisClusterConnection redisClusterConnection = RedisClusterHandler.getRedisClusterConnection(clusterId);
        try{
            List<String> nodes = new ArrayList<>();
            Iterator<RedisClusterNode> iterator =  redisClusterConnection.clusterGetNodes().iterator();
            while (iterator.hasNext()){
                nodes.add(iterator.next().toString());
            }
            Cluster cluster = clusterRepository.findOne(clusterId);
            Assert.notNull(cluster);
            String clusterNodes = System.currentTimeMillis() + "|" +
                                  cluster.getType() + "|" +
                                  StringUtils.join(nodes.toArray(new String[0]), ",");
            String messageKey = busiCode+"_current_cluster";
            redisClusterConnection.publish(SafeEncoder.encode(messageKey), SafeEncoder.encode(clusterNodes));
            redisClusterConnection.set(SafeEncoder.encode(messageKey), SafeEncoder.encode(clusterNodes));

            List list = jdbcTemplate.queryForList("select 1 from redis_current_cluster where busi_code=?", new Object[]{busiCode});
            if (list.size() > 0){
                jdbcTemplate.update("update redis_current_cluster set cluster_nodes=? where busi_code=?", new
                        Object[]{clusterNodes, busiCode});
            }else{
                jdbcTemplate.update("insert into redis_current_cluster values (?, ?)", new Object[]{busiCode, clusterNodes});
            }

        }finally {
            try {
                ((JedisCluster) redisClusterConnection.getNativeConnection()).close();
            } catch (IOException e) {
            }
        }

    }



}
